/*
 * Copyright (c) 2012-2015 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package com.parsechina.hpush.broker.spi.persistence.redis;

import com.parsechina.hpush.broker.config.MqttConstants;
import com.parsechina.hpush.broker.internal.MessageGUID;
import com.parsechina.hpush.broker.internal.StoredMessage;
import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.spi.IMessagesStore;
import com.parsechina.hpush.broker.spi.persistence.SessionsStoreUtils;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;

/**
 * IMessagesStore implementation backed by Redis.
 *
 * @author linfeng
 */
class RedisMessagesStore implements IMessagesStore {

    private static final Logger LOG = LoggerFactory.getLogger(RedisMessagesStore.class);

    private final RedissonClient redissonClient;


    //maps topic -> guid
    private RedisGuidStore indexStore;
    private ConcurrentMap<String, MessageGUID> mapIndexStore;
    //maps guid to message, it's message store
    private ConcurrentMap<MessageGUID, StoredMessage> persistentMessageStore;


    RedisMessagesStore(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    @Override
    public void initStore() {
        indexStore = new RedisGuidStore(redissonClient);
        persistentMessageStore = redissonClient.getMap(MqttConstants.CACHE_PERSISTED_MESSAGES_KEY);
        mapIndexStore = redissonClient.getMap(MqttConstants.CACHE_RETAINED_KEY);
        indexStore.initGuidIndexStore(mapIndexStore);
    }

    @Override
    public void storeRetained(String topic, MessageGUID guid) {

        MessageGUID oldGuid = mapIndexStore.get(topic);
        if (oldGuid != null) {
            persistentMessageStore.remove(oldGuid);
        }

        mapIndexStore.put(topic, guid);
        indexStore.putGuidIndex(topic, guid);
    }


    @Override
    public Collection<StoredMessage> searchMessage(Subscription condition) {

        List<StoredMessage> results = new ArrayList<>();
        String topic = condition.getTopicFilter();

        if (topic.contains("+")) {

            List<MessageGUID> guidList = indexStore.complexSearchGuidIndex(topic);
            for (MessageGUID guid : guidList) {
                StoredMessage storedMsg = persistentMessageStore.get(guid);
                results.add(storedMsg);
            }
        } else if (topic.charAt(topic.length() - 1) == '#') {

            List<MessageGUID> guidList = indexStore.searchPrefixGuidIndex(condition.getTopicFilter());
            for (MessageGUID guid : guidList) {
                StoredMessage storedMsg = persistentMessageStore.get(guid);
                results.add(storedMsg);
            }
        } else {

            MessageGUID guid = mapIndexStore.get(topic);
            if (guid != null) {
                StoredMessage storedMsg = persistentMessageStore.get(guid);
                results.add(storedMsg);
            }

        }

        return results;

    }

    @Override
    public MessageGUID storePublishForFuture(StoredMessage evt) {

        LOG.debug("storePublishForFuture store evt {}", evt);
        if (evt.getClientId() == null) {
            LOG.error("persisting a message without a clientId, bad programming error msg: {}", evt);
            throw new IllegalArgumentException("\"persisting a message without a clientId, bad programming error");
        }
        MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
        evt.setGuid(guid);
        LOG.debug("storePublishForFuture guid <{}>", guid);
        persistentMessageStore.put(guid, evt);
        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = redissonClient.getMap(SessionsStoreUtils.messageId2GuidsMapName(evt.getClientId()));
        messageIdToGuid.put(evt.getMessageID(), guid);

        return guid;
    }

    @Override
    public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
        List<StoredMessage> ret = new ArrayList<>();
        for (MessageGUID guid : guids) {
            ret.add(persistentMessageStore.get(guid));
        }
        return ret;
    }

    @Override
    public void dropMessagesInSession(String clientID) {
        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = redissonClient.getMap(SessionsStoreUtils.messageId2GuidsMapName(clientID));
        if (messageIdToGuid != null) {
            for (MessageGUID guid : messageIdToGuid.values()) {
                removeStoredMessage(guid);
            }
            messageIdToGuid.clear();
        }
    }

    @Override
    public void removeStoredMessage(MessageGUID guid) {
        //remove only the not retained and no more referenced
        StoredMessage storedMessage = persistentMessageStore.get(guid);
        if (storedMessage != null && !storedMessage.isRetained() && storedMessage.getReferenceCounter() <= 0) {
            if (storedMessage.getReferenceCounter() < 0) {
                LOG.error("we should never have gotten a reference count less than zero");
            }
            LOG.debug("Cleaning not retained message guid {}", guid);
            persistentMessageStore.remove(guid);
        }
    }

    @Override
    public StoredMessage getMessageByGuid(MessageGUID guid) {
        return persistentMessageStore.get(guid);
    }

    @Override
    public void cleanRetained(String topic) {

        MessageGUID guid = mapIndexStore.get(topic);
        if (guid != null) {
            persistentMessageStore.remove(guid);
        }

        mapIndexStore.remove(topic);
        indexStore.removeGuidIndex(topic);
    }

    @Override
    public void incUsageCounter(MessageGUID guid) {
        StoredMessage storedMessage = persistentMessageStore.get(guid);
        storedMessage.incReferenceCounter();
        persistentMessageStore.put(guid, storedMessage);
    }

    @Override
    public void decUsageCounter(MessageGUID guid) {
        StoredMessage storedMessage = persistentMessageStore.get(guid);
        storedMessage.decReferenceCounter();
        persistentMessageStore.put(guid, storedMessage);
    }
}
